package FlinkSQL

import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api.{DataTypes, EnvironmentSettings}
import org.apache.flink.table.api.bridge.scala._
import org.apache.flink.table.descriptors.{Csv, Kafka, Schema}

/**
 *
 * @program: FlinkSQL
 * @author: YCLW058
 * @create: 2021-05-26 9:41
 * @decsription:
 *
 * */

object KafkaPipelineTest {
  def main(args: Array[String]): Unit = {
    //1 env
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    val tableEnv = StreamTableEnvironment.create(env)

    //2 source
    //从kafka读取数据
    tableEnv.connect(new Kafka()
      .version("universal")
      .topic("input_kafka")
      .property("zookeeper.connect", "localhost:2181")
      .property("bootstrap.servers", "localhost:9092")
    )
      .withFormat(new Csv())
      .withSchema(new Schema()
        .field("id", DataTypes.INT())
        .field("name", DataTypes.STRING())
        .field("age", DataTypes.INT())
      ).createTemporaryTable("kafkaInputTable")

    //3 transformation
    val sTable = tableEnv.from("kafkaInputTable")
    val resultTable = tableEnv.sqlQuery(
      """
        |select id,name
        |from kafkaInputTable
        |""".stripMargin)

    //4 sink
    tableEnv.connect(new Kafka()
      .version("universal")
      .topic("output_kafka")
      .property("zookeeper.connect", "localhost:2181")
      .property("bootstrap.servers", "localhost:9092")
    )
      .withFormat(new Csv())
      .withSchema(new Schema()
        .field("id", DataTypes.INT())
        .field("name", DataTypes.STRING())
      ).createTemporaryTable("kafkaOutputTable")

    //4 sink
    resultTable.insertInto("kafkaOutputTable")
    //resultTable.toAppendStream[(Int,String)].print("SQL")
    //5 execute
   // env.execute("kafka")
    tableEnv.execute("kafka")
  }

}
